#![deny(unreachable_pub)]
#![deny(rustdoc::broken_intra_doc_links)]
#![deny(rustdoc::private_intra_doc_links)]
#![deny(rustdoc::invalid_codeblock_attributes)]
#![deny(rustdoc::invalid_rust_codeblocks)]
#![cfg_attr(docsrs, feature(doc_auto_cfg))]

use thiserror::Error;

use futures::stream::Stream;
use tokio::sync::oneshot;
use tracing::{debug, error};

use core::fmt;
use std::collections::HashMap;

use std::iter;
use std::net::SocketAddr;
use std::option;
use std::pin::Pin;
use std::slice;
use std::str::{self, FromStr};
use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
use std::task::{Context, Poll};
use tokio::io::ErrorKind;
use url::{Host, Url};

use crate::error::{ConnectError, ConnectErrorKind};
use bytes::Bytes;
use serde::{Deserialize, Serialize};
use serde_repr::{Deserialize_repr, Serialize_repr};
use tokio::io;
use tokio::sync::mpsc;
use tokio::task;
pub type Error = Box<dyn std::error::Error + Send + Sync + 'static>;

pub mod handler;
pub(crate) use handler::*;

const VERSION: &str = env!("CARGO_PKG_VERSION");
const LANG: &str = "rust";
const MAX_PENDING_PINGS: usize = 2;
const MULTIPLEXER_SID: u64 = 0;

/// A re-export of the `rustls` crate used in this crate,
/// for use in cases where manual client configurations
/// must be provided using `Options::tls_client_config`.
pub use tokio_rustls::rustls;

use connection::State;
use connector::{Connector, ConnectorOptions};
pub use header::{HeaderMap, HeaderName, HeaderValue};
pub use subject::Subject;

mod auth;
pub(crate) mod auth_utils;
pub mod client;
pub mod connection;
mod connector;
pub mod error;
pub mod nats_server;
mod options;
pub use auth::Auth;
pub use client::{
    Client, PublishError, Request, RequestError, RequestErrorKind, Statistics, SubscribeError,
};
pub use options::{AuthError, ConnectOptions};

mod crypto;

pub mod header;
pub mod jetstream;
pub mod message;
#[cfg(feature = "service")]
pub mod service;
pub mod status;
pub mod subject;
mod tls;

pub use message::Message;
pub use status::StatusCode;

/// Information sent by the server back to this client
/// during initial connection, and possibly again later.
#[derive(Debug, Deserialize, Default, Clone, Eq, PartialEq)]
pub struct ServerInfo {
    /// The unique identifier of the NATS server.
    #[serde(default)]
    pub server_id: String,
    /// Generated Server Name.
    #[serde(default)]
    pub server_name: String,
    /// The host specified in the cluster parameter/options.
    #[serde(default)]
    pub host: String,
    /// The port number specified in the cluster parameter/options.
    #[serde(default)]
    pub port: u16,
    /// The version of the NATS server.
    #[serde(default)]
    pub version: String,
    /// If this is set, then the server should try to authenticate upon
    /// connect.
    #[serde(default)]
    pub auth_required: bool,
    /// If this is set, then the server must authenticate using TLS.
    #[serde(default)]
    pub tls_required: bool,
    /// Maximum payload size that the server will accept.
    #[serde(default)]
    pub max_payload: usize,
    /// The protocol version in use.
    #[serde(default)]
    pub proto: i8,
    /// The server-assigned client ID. This may change during reconnection.
    #[serde(default)]
    pub client_id: u64,
    /// The version of golang the NATS server was built with.
    #[serde(default)]
    pub go: String,
    /// The nonce used for nkeys.
    #[serde(default)]
    pub nonce: String,
    /// A list of server urls that a client can connect to.
    #[serde(default)]
    pub connect_urls: Vec<String>,
    /// The client IP as known by the server.
    #[serde(default)]
    pub client_ip: String,
    /// Whether the server supports headers.
    #[serde(default)]
    pub headers: bool,
    /// Whether server goes into lame duck mode.
    #[serde(default, rename = "ldm")]
    pub lame_duck_mode: bool,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum ServerOp {
    Ok,
    Info(Box<ServerInfo>),
    Ping,
    Pong,
    Error(ServerError),
    Message {
        sid: u64,
        subject: Subject,
        reply: Option<Subject>,
        payload: Bytes,
        headers: Option<HeaderMap>,
        status: Option<StatusCode>,
        description: Option<String>,
        length: usize,
    },
}

/// `PublishMessage` represents a message being published
#[derive(Debug)]
pub struct PublishMessage {
    pub subject: Subject,
    pub payload: Bytes,
    pub reply: Option<Subject>,
    pub headers: Option<HeaderMap>,
}

/// `Command` represents all commands that a [`Client`] can handle
#[derive(Debug)]
pub(crate) enum Command {
    Publish(PublishMessage),
    Request {
        subject: Subject,
        payload: Bytes,
        respond: Subject,
        headers: Option<HeaderMap>,
        sender: oneshot::Sender<Message>,
    },
    Subscribe {
        sid: u64,
        subject: Subject,
        queue_group: Option<String>,
        sender: mpsc::Sender<Message>,
    },
    Unsubscribe {
        sid: u64,
        max: Option<u64>,
    },
    Flush {
        observer: oneshot::Sender<()>,
    },
    Drain {
        sid: Option<u64>,
    },
    Reconnect,
}

/// `ClientOp` represents all actions of `Client`.
#[derive(Debug)]
pub(crate) enum ClientOp {
    Publish {
        subject: Subject,
        payload: Bytes,
        respond: Option<Subject>,
        headers: Option<HeaderMap>,
    },
    Subscribe {
        sid: u64,
        subject: Subject,
        queue_group: Option<String>,
    },
    Unsubscribe {
        sid: u64,
        max: Option<u64>,
    },
    Ping,
    Pong,
    Connect(ConnectInfo),
}

#[derive(Debug)]
struct Subscription {
    subject: Subject,
    sender: mpsc::Sender<Message>,
    queue_group: Option<String>,
    delivered: u64,
    max: Option<u64>,
    is_draining: bool,
}

#[derive(Debug)]
struct Multiplexer {
    subject: Subject,
    prefix: Subject,
    senders: HashMap<String, oneshot::Sender<Message>>,
}

/// Connects to NATS with specified options.
///
/// It is generally advised to use [ConnectOptions] instead, as it provides a builder for whole
/// configuration.
///
/// # Examples
/// ```
/// # #[tokio::main]
/// # async fn main() ->  Result<(), async_nats::Error> {
/// let mut nc =
///     async_nats::connect_with_options("demo.nats.io", async_nats::ConnectOptions::new()).await?;
/// nc.publish("test", "data".into()).await?;
/// # Ok(())
/// # }
/// ```
pub async fn connect_with_options<A: ToServerAddrs>(
    addrs: A,
    options: ConnectOptions,
) -> Result<Client, ConnectError> {
    let ping_period = options.ping_interval;

    let (events_tx, mut events_rx) = mpsc::channel(128);
    let (state_tx, state_rx) = tokio::sync::watch::channel(State::Pending);
    // We're setting it to the default server payload size.
    let max_payload = Arc::new(AtomicUsize::new(1024 * 1024));
    let statistics = Arc::new(Statistics::default());

    let mut connector = Connector::new(
        addrs,
        ConnectorOptions {
            tls_required: options.tls_required,
            certificates: options.certificates,
            client_key: options.client_key,
            client_cert: options.client_cert,
            tls_client_config: options.tls_client_config,
            tls_first: options.tls_first,
            auth: options.auth,
            no_echo: options.no_echo,
            connection_timeout: options.connection_timeout,
            name: options.name,
            ignore_discovered_servers: options.ignore_discovered_servers,
            retain_servers_order: options.retain_servers_order,
            read_buffer_capacity: options.read_buffer_capacity,
            reconnect_delay_callback: options.reconnect_delay_callback,
            auth_callback: options.auth_callback,
            max_reconnects: options.max_reconnects,
        },
        events_tx,
        state_tx,
        max_payload.clone(),
        statistics.clone(),
    )
    .map_err(|err| ConnectError::with_source(ConnectErrorKind::ServerParse, err))?;

    let mut info: ServerInfo = Default::default();
    let mut connection = None;

    if !options.retry_on_initial_connect {
        debug!("retry on initial connect failure is disabled");
        let (info_ok, connection_ok) = connector.try_connect().await?;
        connection = Some(connection_ok);
        info = info_ok;
    }

    let (info_sender, info_watcher) = tokio::sync::watch::channel(info.clone());
    let (sender, mut receiver) = mpsc::channel(options.sender_capacity);

    let client = Client::new(
        info_watcher,
        state_rx,
        sender,
        options.subscription_capacity,
        options.inbox_prefix,
        options.request_timeout,
        max_payload,
        statistics,
    );

    //接收事件
    task::spawn(async move {
        while let Some(event) = events_rx.recv().await {
            tracing::info!("event: {}", event);
            if let Some(event_callback) = &options.event_callback {
                event_callback.call(event).await;
            }
        }
    });

    task::spawn(async move {
        if connection.is_none() && options.retry_on_initial_connect {
            let (info, connection_ok) = match connector.connect().await {
                Ok((info, connection)) => (info, connection),
                Err(err) => {
                    error!("connection closed: {}", err);
                    return;
                }
            };
            info_sender.send(info).ok();
            connection = Some(connection_ok);
        }
        let connection = connection.unwrap();

        //将连接掉到 handlers中
        let mut connection_handler =
            ConnectionHandler::new(connection, connector, info_sender, ping_period);
        connection_handler.process(&mut receiver).await
    });

    Ok(client)
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Event {
    Connected,
    Disconnected,
    LameDuckMode,
    Draining,
    Closed,
    SlowConsumer(u64),
    ServerError(ServerError),
    ClientError(ClientError),
}

impl fmt::Display for Event {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Event::Connected => write!(f, "connected"),
            Event::Disconnected => write!(f, "disconnected"),
            Event::LameDuckMode => write!(f, "lame duck mode detected"),
            Event::Draining => write!(f, "draining"),
            Event::Closed => write!(f, "closed"),
            Event::SlowConsumer(sid) => write!(f, "slow consumers for subscription {sid}"),
            Event::ServerError(err) => write!(f, "server error: {err}"),
            Event::ClientError(err) => write!(f, "client error: {err}"),
        }
    }
}

/// Connects to NATS with default config.
///
/// Returns cloneable [Client].
///
/// To have customized NATS connection, check [ConnectOptions].
///
/// # Examples
///
/// ## Single URL
/// ```
/// # #[tokio::main]
/// # async fn main() ->  Result<(), async_nats::Error> {
/// let mut nc = async_nats::connect("demo.nats.io").await?;
/// nc.publish("test", "data".into()).await?;
/// # Ok(())
/// # }
/// ```
///
/// ## Connect with [Vec] of [ServerAddr].
/// ```no_run
/// #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// use async_nats::ServerAddr;
/// let client = async_nats::connect(vec![
///     "demo.nats.io".parse::<ServerAddr>()?,
///     "other.nats.io".parse::<ServerAddr>()?,
/// ])
/// .await
/// .unwrap();
/// # Ok(())
/// # }
/// ```
///
/// ## with [Vec], but parse URLs inside [crate::connect()]
/// ```no_run
/// #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// use async_nats::ServerAddr;
/// let servers = vec!["demo.nats.io", "other.nats.io"];
/// let client = async_nats::connect(
///     servers
///         .iter()
///         .map(|url| url.parse())
///         .collect::<Result<Vec<ServerAddr>, _>>()?,
/// )
/// .await?;
/// # Ok(())
/// # }
/// ```
///
///
/// ## with slice.
/// ```no_run
/// #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// use async_nats::ServerAddr;
/// let client = async_nats::connect(
///    [
///        "demo.nats.io".parse::<ServerAddr>()?,
///        "other.nats.io".parse::<ServerAddr>()?,
///    ]
///    .as_slice(),
/// )
/// .await?;
/// # Ok(())
/// # }
pub async fn connect<A: ToServerAddrs>(addrs: A) -> Result<Client, ConnectError> {
    connect_with_options(addrs, ConnectOptions::default()).await
}

/// Retrieves messages from given `subscription` created by [Client::subscribe].
///
/// Implements [futures::stream::Stream] for ergonomic async message processing.
///
/// # Examples
/// ```
/// # #[tokio::main]
/// # async fn main() ->  Result<(), async_nats::Error> {
/// let mut nc = async_nats::connect("demo.nats.io").await?;
/// # nc.publish("test", "data".into()).await?;
/// # Ok(())
/// # }
/// ```
#[derive(Debug)]
pub struct Subscriber {
    sid: u64,
    receiver: mpsc::Receiver<Message>,
    sender: mpsc::Sender<Command>,
}

impl Subscriber {
    fn new(
        sid: u64,
        sender: mpsc::Sender<Command>,
        receiver: mpsc::Receiver<Message>,
    ) -> Subscriber {
        Subscriber {
            sid,
            sender,
            receiver,
        }
    }

    /// Unsubscribes from subscription, draining all remaining messages.
    ///
    /// # Examples
    /// ```
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// let client = async_nats::connect("demo.nats.io").await?;
    ///
    /// let mut subscriber = client.subscribe("foo").await?;
    ///
    /// subscriber.unsubscribe().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn unsubscribe(&mut self) -> Result<(), UnsubscribeError> {
        self.sender
            .send(Command::Unsubscribe {
                sid: self.sid,
                max: None,
            })
            .await?;
        self.receiver.close();
        Ok(())
    }

    /// Unsubscribes from subscription after reaching given number of messages.
    /// This is the total number of messages received by this subscription in it's whole
    /// lifespan. If it already reached or surpassed the passed value, it will immediately stop.
    ///
    /// # Examples
    /// ```
    /// # use futures::StreamExt;
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// let client = async_nats::connect("demo.nats.io").await?;
    ///
    /// let mut subscriber = client.subscribe("test").await?;
    /// subscriber.unsubscribe_after(3).await?;
    ///
    /// for _ in 0..3 {
    ///     client.publish("test", "data".into()).await?;
    /// }
    ///
    /// while let Some(message) = subscriber.next().await {
    ///     println!("message received: {:?}", message);
    /// }
    /// println!("no more messages, unsubscribed");
    /// # Ok(())
    /// # }
    /// ```
    pub async fn unsubscribe_after(&mut self, unsub_after: u64) -> Result<(), UnsubscribeError> {
        self.sender
            .send(Command::Unsubscribe {
                sid: self.sid,
                max: Some(unsub_after),
            })
            .await?;
        Ok(())
    }

    /// Unsubscribes immediately but leaves the subscription open to allow any in-flight messages
    /// on the subscription to be delivered. The stream will be closed after any remaining messages
    /// are delivered
    ///
    /// # Examples
    /// ```no_run
    /// # use futures::StreamExt;
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// let client = async_nats::connect("demo.nats.io").await?;
    ///
    /// let mut subscriber = client.subscribe("test").await?;
    ///
    /// tokio::spawn({
    ///     let task_client = client.clone();
    ///     async move {
    ///         loop {
    ///             _ = task_client.publish("test", "data".into()).await;
    ///         }
    ///     }
    /// });
    ///
    /// client.flush().await?;
    /// subscriber.drain().await?;
    ///
    /// while let Some(message) = subscriber.next().await {
    ///     println!("message received: {:?}", message);
    /// }
    /// println!("no more messages, unsubscribed");
    /// # Ok(())
    /// # }
    /// ```
    pub async fn drain(&mut self) -> Result<(), UnsubscribeError> {
        self.sender
            .send(Command::Drain {
                sid: Some(self.sid),
            })
            .await?;

        Ok(())
    }
}

#[derive(Error, Debug, PartialEq)]
#[error("failed to send unsubscribe")]
pub struct UnsubscribeError(String);

impl From<tokio::sync::mpsc::error::SendError<Command>> for UnsubscribeError {
    fn from(err: tokio::sync::mpsc::error::SendError<Command>) -> Self {
        UnsubscribeError(err.to_string())
    }
}

impl Drop for Subscriber {
    fn drop(&mut self) {
        self.receiver.close();
        tokio::spawn({
            let sender = self.sender.clone();
            let sid = self.sid;
            async move {
                sender
                    .send(Command::Unsubscribe { sid, max: None })
                    .await
                    .ok();
            }
        });
    }
}

impl Stream for Subscriber {
    type Item = Message;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.receiver.poll_recv(cx)
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum CallbackError {
    Client(ClientError),
    Server(ServerError),
}
impl std::fmt::Display for CallbackError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Client(error) => write!(f, "{error}"),
            Self::Server(error) => write!(f, "{error}"),
        }
    }
}

impl From<ServerError> for CallbackError {
    fn from(server_error: ServerError) -> Self {
        CallbackError::Server(server_error)
    }
}

impl From<ClientError> for CallbackError {
    fn from(client_error: ClientError) -> Self {
        CallbackError::Client(client_error)
    }
}

#[derive(Clone, Debug, Eq, PartialEq, Error)]
pub enum ServerError {
    AuthorizationViolation,
    SlowConsumer(u64),
    Other(String),
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ClientError {
    Other(String),
    MaxReconnects,
}
impl std::fmt::Display for ClientError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Other(error) => write!(f, "nats: {error}"),
            Self::MaxReconnects => write!(f, "nats: max reconnects reached"),
        }
    }
}

impl ServerError {
    fn new(error: String) -> ServerError {
        match error.to_lowercase().as_str() {
            "authorization violation" => ServerError::AuthorizationViolation,
            // error messages can contain case-sensitive values which should be preserved
            _ => ServerError::Other(error),
        }
    }
}

impl std::fmt::Display for ServerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::AuthorizationViolation => write!(f, "nats: authorization violation"),
            Self::SlowConsumer(sid) => write!(f, "nats: subscription {sid} is a slow consumer"),
            Self::Other(error) => write!(f, "nats: {error}"),
        }
    }
}

/// Info to construct a CONNECT message.
#[derive(Clone, Debug, Serialize)]
pub struct ConnectInfo {
    /// Turns on +OK protocol acknowledgments.
    pub verbose: bool,

    /// Turns on additional strict format checking, e.g. for properly formed
    /// subjects.
    pub pedantic: bool,

    /// User's JWT.
    #[serde(rename = "jwt")]
    pub user_jwt: Option<String>,

    /// Public nkey.
    pub nkey: Option<String>,

    /// Signed nonce, encoded to Base64URL.
    #[serde(rename = "sig")]
    pub signature: Option<String>,

    /// Optional client name.
    pub name: Option<String>,

    /// If set to `true`, the server (version 1.2.0+) will not send originating
    /// messages from this connection to its own subscriptions. Clients should
    /// set this to `true` only for server supporting this feature, which is
    /// when proto in the INFO protocol is set to at least 1.
    pub echo: bool,

    /// The implementation language of the client.
    pub lang: String,

    /// The version of the client.
    pub version: String,

    /// Sending 0 (or absent) indicates client supports original protocol.
    /// Sending 1 indicates that the client supports dynamic reconfiguration
    /// of cluster topology changes by asynchronously receiving INFO messages
    /// with known servers it can reconnect to.
    pub protocol: Protocol,

    /// Indicates whether the client requires an SSL connection.
    pub tls_required: bool,

    /// Connection username (if `auth_required` is set)
    pub user: Option<String>,

    /// Connection password (if auth_required is set)
    pub pass: Option<String>,

    /// Client authorization token (if auth_required is set)
    pub auth_token: Option<String>,

    /// Whether the client supports the usage of headers.
    pub headers: bool,

    /// Whether the client supports no_responders.
    pub no_responders: bool,
}

/// Protocol version used by the client.
#[derive(Serialize_repr, Deserialize_repr, PartialEq, Eq, Debug, Clone, Copy)]
#[repr(u8)]
pub enum Protocol {
    /// Original protocol.
    Original = 0,
    /// Protocol with dynamic reconfiguration of cluster and lame duck mode functionality.
    Dynamic = 1,
}

/// Address of a NATS server.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct ServerAddr(Url);

impl FromStr for ServerAddr {
    type Err = io::Error;

    /// Parse an address of a NATS server.
    ///
    /// If not stated explicitly the `nats://` schema and port `4222` is assumed.
    fn from_str(input: &str) -> Result<Self, Self::Err> {
        let url: Url = if input.contains("://") {
            input.parse()
        } else {
            format!("nats://{input}").parse()
        }
        .map_err(|e| {
            io::Error::new(
                ErrorKind::InvalidInput,
                format!("NATS server URL is invalid: {e}"),
            )
        })?;

        Self::from_url(url)
    }
}

impl ServerAddr {
    /// Check if the URL is a valid NATS server address.
    pub fn from_url(url: Url) -> io::Result<Self> {
        if url.scheme() != "nats"
            && url.scheme() != "tls"
            && url.scheme() != "ws"
            && url.scheme() != "wss"
        {
            return Err(std::io::Error::new(
                ErrorKind::InvalidInput,
                format!("invalid scheme for NATS server URL: {}", url.scheme()),
            ));
        }

        Ok(Self(url))
    }

    /// Turn the server address into a standard URL.
    pub fn into_inner(self) -> Url {
        self.0
    }

    /// Returns if tls is required by the client for this server.
    pub fn tls_required(&self) -> bool {
        self.0.scheme() == "tls"
    }

    /// Returns if the server url had embedded username and password.
    pub fn has_user_pass(&self) -> bool {
        self.0.username() != ""
    }

    pub fn scheme(&self) -> &str {
        self.0.scheme()
    }

    /// Returns the host.
    pub fn host(&self) -> &str {
        match self.0.host() {
            Some(Host::Domain(_)) | Some(Host::Ipv4 { .. }) => self.0.host_str().unwrap(),
            // `host_str()` for Ipv6 includes the []s
            Some(Host::Ipv6 { .. }) => {
                let host = self.0.host_str().unwrap();
                &host[1..host.len() - 1]
            }
            None => "",
        }
    }

    pub fn is_websocket(&self) -> bool {
        self.0.scheme() == "ws" || self.0.scheme() == "wss"
    }

    /// Returns the port.
    pub fn port(&self) -> u16 {
        self.0.port().unwrap_or(4222)
    }

    /// Returns the optional username in the url.
    pub fn username(&self) -> Option<&str> {
        let user = self.0.username();
        if user.is_empty() {
            None
        } else {
            Some(user)
        }
    }

    /// Returns the optional password in the url.
    pub fn password(&self) -> Option<&str> {
        self.0.password()
    }

    /// Return the sockets from resolving the server address.
    pub async fn socket_addrs(&self) -> io::Result<impl Iterator<Item = SocketAddr> + '_> {
        tokio::net::lookup_host((self.host(), self.port())).await
    }
}

/// Capability to convert into a list of NATS server addresses.
///
/// There are several implementations ensuring the easy passing of one or more server addresses to
/// functions like [`crate::connect()`].
pub trait ToServerAddrs {
    /// Returned iterator over socket addresses which this type may correspond
    /// to.
    type Iter: Iterator<Item = ServerAddr>;

    fn to_server_addrs(&self) -> io::Result<Self::Iter>;
}

impl ToServerAddrs for ServerAddr {
    type Iter = option::IntoIter<ServerAddr>;
    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        Ok(Some(self.clone()).into_iter())
    }
}

impl ToServerAddrs for str {
    type Iter = option::IntoIter<ServerAddr>;
    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        self.parse::<ServerAddr>()
            .map(|addr| Some(addr).into_iter())
    }
}

impl ToServerAddrs for String {
    type Iter = option::IntoIter<ServerAddr>;
    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        (**self).to_server_addrs()
    }
}

impl<T: AsRef<str>> ToServerAddrs for [T] {
    type Iter = std::vec::IntoIter<ServerAddr>;
    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        self.iter()
            .map(AsRef::as_ref)
            .map(str::parse)
            .collect::<io::Result<_>>()
            .map(Vec::into_iter)
    }
}

impl<T: AsRef<str>> ToServerAddrs for Vec<T> {
    type Iter = std::vec::IntoIter<ServerAddr>;
    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        self.as_slice().to_server_addrs()
    }
}

impl<'a> ToServerAddrs for &'a [ServerAddr] {
    type Iter = iter::Cloned<slice::Iter<'a, ServerAddr>>;

    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        Ok(self.iter().cloned())
    }
}

impl ToServerAddrs for Vec<ServerAddr> {
    type Iter = std::vec::IntoIter<ServerAddr>;

    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        Ok(self.clone().into_iter())
    }
}

impl<T: ToServerAddrs + ?Sized> ToServerAddrs for &T {
    type Iter = T::Iter;
    fn to_server_addrs(&self) -> io::Result<Self::Iter> {
        (**self).to_server_addrs()
    }
}

pub(crate) fn is_valid_subject<T: AsRef<str>>(subject: T) -> bool {
    let subject_str = subject.as_ref();
    !subject_str.starts_with('.')
        && !subject_str.ends_with('.')
        && subject_str.bytes().all(|c| !c.is_ascii_whitespace())
}
macro_rules! from_with_timeout {
    ($t:ty, $k:ty, $origin: ty, $origin_kind: ty) => {
        impl From<$origin> for $t {
            fn from(err: $origin) -> Self {
                match err.kind() {
                    <$origin_kind>::TimedOut => Self::new(<$k>::TimedOut),
                    _ => Self::with_source(<$k>::Other, err),
                }
            }
        }
    };
}
pub(crate) use from_with_timeout;
